package mq

import (
	"Metro/config"
	"fmt"
	"github.com/vmihailenco/msgpack/v5"
	"jialiking.com/gaogang/common/utils"
	"math/rand"
	"runtime"
	"strconv"
	"time"

	"github.com/streadway/amqp"
)

type Config struct {
	ChannelMax int
	Health     time.Duration
	Heartbeat  time.Duration
}

type Channel struct {
	ChId int64
	Ch   *amqp.Channel
	Con  *amqp.Connection
}

type RabbitPool struct {
	url        string
	channelMax int
	health     time.Duration
	heartbeat  time.Duration
	channels   chan *Channel
}

var Pool *RabbitPool

func init() {
	Pool = New(config.AMQP, Config{
		ChannelMax: runtime.NumCPU(),
		Health:     30 * time.Second,
	})
}

func New(url string, config Config) *RabbitPool {
	r := &RabbitPool{
		url:        url,
		channelMax: config.ChannelMax,
		health:     config.Health,
		heartbeat:  config.Heartbeat,
		channels:   make(chan *Channel, config.ChannelMax),
	}

	for channelId := 0; channelId < r.channelMax; channelId++ {
		channel, err := r.newChannel(int64(channelId))
		if err != nil {
			panic("Create channel pool error")
		}
		r.channels <- channel
	}
	go r.healthCheck()
	return r
}

func (r *RabbitPool) newChannel(channelID int64) (*Channel, error) {
	conn, err := amqp.DialConfig(
		r.url, amqp.Config{
			Heartbeat: r.heartbeat,
		},
	)
	if err != nil {
		return nil, err
	}
	channel, err := conn.Channel()
	if err != nil {
		_ = conn.Close()
		return nil, err
	}
	rc := &Channel{
		ChId: channelID,
		Con:  conn,
		Ch:   channel,
	}
	return rc, nil
}

func (r *RabbitPool) Get() *Channel {
	return <-r.channels
}

func (r *RabbitPool) Put(ch *Channel) {
	r.channels <- ch
}

func (r *RabbitPool) healthCheck() {
	ticker := time.NewTicker(r.health)
	for {
		<-ticker.C
		for i := 0; i < r.channelMax; i++ {
			channel := r.Get()
			if channel.Con.IsClosed() {
				_ = channel.Con.Close()
				newChannel, err := r.newChannel(channel.ChId)
				if err != nil {
					utils.ErrorLog("Create channel error in healthCheck, %s", err.Error())
					r.Put(channel)
				} else {
					r.Put(newChannel)
				}
			} else {
				r.Put(channel)
			}
		}
	}
}

func SendMQ(funcStr string, args []interface{}, kwargs map[string]interface{}) {
	utils.InfoLog("SendMQ: %s, args: %v, kwargs: %v", funcStr, args, kwargs)
	data := []interface{}{
		args,
		kwargs,
		map[string]interface{}{
			"callbacks": nil,
			"errbacks":  nil,
			"chain":     nil,
			"chord":     nil,
		},
	}
	body, err := msgpack.Marshal(data)
	if err != nil {
		panic("msgpack decode error")
	}

	id := strconv.Itoa(rand.Int())
	if err != nil {
		utils.ErrorLog("rabbitmq get channel error: %s", err.Error())
		return
	}
	ch := Pool.Get()
	defer Pool.Put(ch)
	err = ch.Ch.Publish(
		"",       // exchange
		"celery", // routing key
		false,    // mandatory
		false,    // immediate
		amqp.Publishing{
			Headers: amqp.Table{
				"id":            id,
				"root_id":       id,
				"ignore_result": true,
				"task":          funcStr,
				"argsrepr":      fmt.Sprintf("%v", args),
				"kwargsrepr":    fmt.Sprintf("%v", kwargs),
				"retries":       0,
				"lang":          "go",
				"origin":        "go_client",
			},
			ContentType:     "application/x-msgpack",
			ContentEncoding: "binary",
			ReplyTo:         "",
			Body:            body,
		})
}
